/**
 * @AccessTcpConnector.java  COPYRIGHT xiaojian
 * 
 * Created on 2017年7月6日 下午3:38:09
 */
package xj.toolkit.access.connector;

import java.net.InetSocketAddress;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Preconditions;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import xj.toolkit.netty.client.TCPConnector;
import xj.toolkit.netty.initializer.TcpClientInitializer;
import xj.toolkit.access.msg.AccessMessage;
import xj.toolkit.base.Sender;

/**
 * Access实现的一个 tcp 连接器。
 * 
 * @author <a href="mailto:cxj2000@gmail.com">xiaojian.cao</a>
 *
 */
public class AccessTcpConnector implements Sender<AccessMessage> {
	private static final Logger logger = LoggerFactory.getLogger(TCPConnector.class);

	private ScheduledExecutorService exec = Executors.newSingleThreadScheduledExecutor();

	private String name = "connector";
	private String destIp = null;
	private int destPort = -1;
	private String localAddress = "0.0.0.0";
	private int localPort;
	private Channel channel = null;
	private TcpClientInitializer tcpClientInitializer;
	private Bootstrap bootstrap = new Bootstrap();
	private NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
	private long retryTimeout = 1000;
	
	public String getDestIp() {
		return destIp;
	}

	public void setDestIp(String destIp) {
		this.destIp = destIp;
	}

	public int getDestPort() {
		return destPort;
	}

	public void setDestPort(int destPort) {
		this.destPort = destPort;
	}

	public TcpClientInitializer getTcpClientInitializer() {
		return tcpClientInitializer;
	}

	public void setTcpClientInitializer(TcpClientInitializer tcpClientInitializer) {
		this.tcpClientInitializer = tcpClientInitializer;
	}

	public long getRetryTimeout() {
		return retryTimeout;
	}

	public void setRetryTimeout(long retryTimeout) {
		this.retryTimeout = retryTimeout;
	}
	
	public String getLocalAddress() {
		return localAddress;
	}

	public void setLocalAddress(String localAddress) {
		this.localAddress = localAddress;
	}

	public int getLocalPort() {
		return localPort;
	}

	public void setLocalPort(int localPort) {
		this.localPort = localPort;
	}

	@ChannelHandler.Sharable
	private class MonitorHandler extends ChannelInboundHandlerAdapter {

		@Override
		public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
			if (!exec.isShutdown()) {
				exec.submit(new Runnable() {

					public void run() {
						onChannelDisconnected(ctx.channel());
					}
				});
			}
		}
	}

	private void doConnect() {
		if (null == destIp || destIp.equals("")) {
			logger.warn(getDesc() + " destIp is null, disable this connector.");
			return;
		}

		ChannelFuture future = bootstrap.connect();

		future.addListener(new ChannelFutureListener() {

			public void operationComplete(final ChannelFuture future) throws Exception {
				exec.submit(new Runnable() {
					public void run() {
						onConnectComplete(future);
					}
				});
			}
		});
	}

	private void onConnectComplete(ChannelFuture future) {
		if (!future.isSuccess()) {
			logger.info(getDesc() + " connect [" + this.destIp + ":" + this.destPort + "] failed, retry...");
			exec.schedule(new Runnable() {

				public void run() {
					doConnect();
				}
			}, retryTimeout, TimeUnit.MILLISECONDS);
		} else {
			// 连接成功
			InetSocketAddress address = (InetSocketAddress) future.channel().localAddress();
			localAddress = address.getAddress().getHostAddress();
			localPort = address.getPort();
		}
	}

	private void onChannelDisconnected(Channel channel) {
		if (logger.isInfoEnabled()) {
			logger.info(getDesc() + " channel : " + channel + "closed, retry connect...");
		}
		exec.schedule(new Runnable() {

			public void run() {
				doConnect();
			}
		}, retryTimeout, TimeUnit.MILLISECONDS);
	}

	public AccessTcpConnector() {
	}

	public AccessTcpConnector(String name) {
		this.name = name;
	}

	public void start() {
		bootstrap.group(eventLoopGroup)
			.channel(NioSocketChannel.class)
			.localAddress(localAddress, localPort)
			.remoteAddress(destIp, destPort);
		
		tcpClientInitializer.addOtherHandler(new MonitorHandler());
		bootstrap.handler(tcpClientInitializer);
		doConnect();
	}

	public void stop() {
		this.exec.shutdownNow();
		this.eventLoopGroup.shutdownGracefully();
		if (this.channel != null) {
			this.channel.close();
		}
	}

	private String getDesc() {
		return "[" + name + "]";
	}

	public boolean isActive() {
		return channel != null && channel.isActive();
	}


	@Override
	public void send(AccessMessage target) {
		Preconditions.checkArgument(target != null);
		if (channel != null && channel.isActive() && channel.isWritable()) {
			channel.writeAndFlush(target);
			return;
		}

		throw new IllegalStateException("Channel is NULL or INACTIVE or CANNOT WRITE.");
	}
	
	public void send(Object req, AccessMessage target) {
		this.send(target);
	}
}
